package com.xkt.service;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.UUID;

/**
 * Description:
 *
 * @Create 2023/12/14 23:46
 * @Version 1.0
 */
@Component
@Slf4j
public class QueueProducer {

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    public void sendQueue(String topic, Object msgContent) {

        String obj2String = JSON.toJSONString(msgContent);

        log.info("kafka service 准备发送消息为：{}", obj2String);

        //发送消息
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, UUID.randomUUID().toString(), obj2String);
        future.addCallback(new ListenableFutureCallback() {
            @Override
            public void onSuccess(Object result) {
                log.info("[kafka service-生产成功]topic：{}，结果{}", topic, result.toString());
            }
            //消息发送失败
            @Override
            public void onFailure(Throwable throwable) {
                //发送失败的处理，本处只是记录了错误日志,可结合实际业务做处理
                log.info("[kafka service-生产失败]topic：{}，失败原因{}", topic, throwable.getMessage());
            }
        });
    }
}

